package com.tyc.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;

/**
 * 配置生产者提高吞吐量
 * 保证数据有序：
 *      单个分区
 *      开启幂等，配置队列缓存数据小于等于5
 *      未开启幂等，配置队列缓存为1
 */
public class CustomProducerParams {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // 连接地址 多个用逗号隔开
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092,127.0.0.1:9093");
        // 指定序列化类型
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 缓冲区大小32M
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");
        // 批次大小16K
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        // 延时
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1");
        // 压缩
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
        /**
         * 默认是all,与-1一样
         * ack 0（只发生产很少使用）
         * 1（leader收到传输普通日志允许丢失）
         * -1（leader与ISR列表都收到,Leader维护,分区副本大于等于2，ISR应答的最小副本数量大于等于2）可能会有重复数据
         * 完全可靠=ack=-1 + 副本数>=2 + isr > 2 ===> 会有数据重复
         * 解决数据重复：幂等性(pid(每次重启加一),分区号，序列化号)(默认打开)+事务
         * 保证数据有序：没有幂等性的话，设置 inflight=1   有幂等性的话，inflight<=5即可，缓存按序落盘，若数据没来则等待
         *
         */
        properties.setProperty(ProducerConfig.ACKS_CONFIG,"-1");
        //重试次数 默认int最大值
        properties.setProperty(ProducerConfig.RETRIES_CONFIG,"3");
        //指定事务ID 开启事务必须指定id，且ack必须是ALL
        properties.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString());
        //创建生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 开始事务保证精确一次
        kafkaProducer.initTransactions();
        kafkaProducer.beginTransaction();
        try {
            for (int i = 0; i < 3; i++) {
                kafkaProducer.send(new ProducerRecord<>("first", i+"hello"), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if(null == e){
                            System.out.println(String.format("主题：%s,分区：%s",recordMetadata.topic(),recordMetadata.partition()));
                        }
                    }
                });
            }
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            kafkaProducer.abortTransaction();
            e.printStackTrace();
        } finally {
            kafkaProducer.close();
        }
    }
}
